package util

import (
	"common/logging"
	"errors"
	"iot/iot_app_server/proto"
	"sync"
	"time"
)

var MsgManagerInstance = NewMsgManager()

type UpstramMsg struct {
	Topic   string
	Para    string
	Payload string
}

type MsgItem struct {
	MsgId   string
	MsgChan chan *MSGBUS_DOWNSTREAM.MqttPub
}

type MsgManager struct {
	msgMap map[string]*MsgItem
	mutex  *sync.Mutex
}

func NewMsgManager() *MsgManager {
	return &MsgManager{
		msgMap: make(map[string]*MsgItem),
		mutex:  &sync.Mutex{},
	}
}

func (m *MsgManager) AddMsg(msg *MsgItem) error {
	if msg == nil || msg.MsgId == "" {
		logging.Error("MsgManager AddMsg invalid msg:%+v", msg)
		return errors.New("AddMsg invalid msg")
	}

	m.mutex.Lock()
	defer m.mutex.Unlock()
	if _, exist := m.msgMap[msg.MsgId]; exist {
		logging.Error("AddMsg msg exist,msg=%+v", msg)
		return errors.New("AddMsg msg exist")
	}
	m.msgMap[msg.MsgId] = msg
	return nil
}

func (m *MsgManager) GetMsg(msgid string) *MsgItem {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	return m.msgMap[msgid]
}

func (m *MsgManager) RemoveMsg(msgid string) error {
	m.mutex.Lock()
	defer m.mutex.Unlock()
	if _, exist := m.msgMap[msgid]; !exist {
		logging.Error("RemoveMsg  msg not exist,msgid=%+v", msgid)
		return errors.New("RemoveMsg  msg not exist")
	}
	delete(m.msgMap, msgid)
	return nil
}

//topic,para,payload,error
type UpMsgHandle func(*MSGBUS_DOWNSTREAM.MqttPub, error)

//注册消息回调，在收到回复或超时时会执行回调函数
func AttachMsgHandle(msgid string, handleFunc UpMsgHandle, tTimeout int, c *chan bool) error {
	item := &MsgItem{
		MsgId:   msgid,
		MsgChan: make(chan *MSGBUS_DOWNSTREAM.MqttPub),
	}
	if err := MsgManagerInstance.AddMsg(item); err != nil {
		logging.Error("AttachMsgHandle AddMsg failed")
		return err
	}

	go func() {
		timer := time.NewTimer(time.Second * time.Duration(tTimeout))
		select {
		case m := <-item.MsgChan:
			if m == nil {
				logging.Error("AttachMsgHandle get invalid upstream msg")
				handleFunc(nil, errors.New("inner error"))
				break
			}
			handleFunc(m, nil)
		case <-timer.C:
			handleFunc(nil, errors.New(ERR_STR_TIMEOUT))
			logging.Debug("AttachMsgHandle msg timeout,msgid=%s", msgid)
		}
		MsgManagerInstance.RemoveMsg(msgid)
		if c != nil {
			select {
			case *c <- true:
			default:
			}
		}
	}()
	return nil
}
